package com.shenmazong.mq.recv;

import com.rabbitmq.client.Channel;
import com.shenmazong.vo.MessageVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.SerializationUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

/**
 * @author 军哥
 * @version 1.0
 * @description: 消费者
 * @date 2022/6/30 11:48
 */

@Component
@Slf4j
public class MessageRecv {

    @Autowired
    RedisTemplate redisTemplate;

    @RabbitListener(queues = "simpleQueue")
    public void onSimpleMessage(Message message, Channel channel) {
        try {
            //--1 接收消息
            byte[] body = message.getBody();
            MessageVo msg = (MessageVo)SerializationUtils.deserialize(body);

            //--2 处理消息

            // 检查重复消息
            Long msgId = msg.getId();
            String key = "msg_"+msgId;
            if(redisTemplate.hasKey(key)) {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
                return;
            }

            //--3
            if(msg.getType().equals(1)) {
                String s = (String)msg.getData();
                log.info("onSimpleMessage:"+s);

                // 设置消息ID，过期时间为5分钟
                redisTemplate.opsForValue().set(key, msgId, 5, TimeUnit.MINUTES);

                // 确认消息：从队列中删除消息
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @RabbitListener(queues = "workQueue")
    public void onWorkMessage(Message message, Channel channel) {
        try {
            //--1
            byte[] body = message.getBody();
            String msg = new String(body);

            //--2 手动确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);

            log.info("onWorkMessage:"+msg);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @RabbitListener(queues = "publishQueueA")
    public void onPublishMessageA(Message message, Channel channel) {
        try {
            //--1
            byte[] body = message.getBody();
            String msg = new String(body);

            //--2 手动确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);

            log.info("onPublishMessageA:"+msg);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @RabbitListener(queues = "publishQueueB")
    public void onPublishMessageB(Message message, Channel channel) {
        try {
            //--1
            byte[] body = message.getBody();
            String msg = new String(body);

            //--2 手动确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);

            log.info("onPublishMessageB:"+msg);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @RabbitListener(queues = "directQueueA")
    public void onDirectMessageA(Message message, Channel channel) {
        try {
            //--1
            byte[] body = message.getBody();
            String msg = new String(body);

            //--2 手动确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);

            log.info("onDirectMessageA:"+msg);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @RabbitListener(queues = "directQueueB")
    public void onDirectQueueB(Message message, Channel channel) {
        try {
            //--1
            byte[] body = message.getBody();
            String msg = new String(body);

            //--2 手动确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);

            log.info("onDirectQueueB:"+msg);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    //=====================================================



    @RabbitListener(queues = "topicQueueA")
    public void onTopicQueueA(Message message, Channel channel) {
        try {
            //--1
            byte[] body = message.getBody();
            String msg = new String(body);

            //--2 手动确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);

            log.info("onTopicQueueA:"+msg);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @RabbitListener(queues = "topicQueueB")
    public void onTopicQueueB(Message message, Channel channel) {
        try {
            //--1
            byte[] body = message.getBody();
            String msg = new String(body);

            //--2 手动确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);

            log.info("onTopicQueueB:"+msg);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}
